package com.shujia.stream

import org.apache.spark.sql.SQLContext
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.{Duration, Durations, StreamingContext}
import org.apache.spark.{SparkConf, SparkContext}

object Demo1WordCOunt {
  def main(args: Array[String]): Unit = {

    /**
      * 创建spark 执行环境
      *
      */

    val conf: SparkConf = new SparkConf()
      .setMaster("local[2]")
      .setAppName("stream")

    val sc = new SparkContext(conf)


    /**
      * 创建 spark streaming 执行环境
      *
      * 指定处理数据的间隔时间
      *
      */

    val ssc = new StreamingContext(sc, Durations.seconds(5))

    //读取实时数据，比如消息队列
    //再linux中安装yum install nc
    //可以通过nc -lk 模拟一个消息队列


    val linesDS: ReceiverInputDStream[String] = ssc.socketTextStream("master", 8888)


    linesDS
      .flatMap(_.split(","))
      .map((_, 1))
      .reduceByKey(_ + _)
      .print()


    //启动spark streaming
    ssc.start()
    ssc.awaitTermination()
    ssc.stop()


  }

}
